package cn.doitedu.day01.rdd

import java.sql.{Connection, DriverManager, ResultSet}

import cn.doitedu.day01.utils.SparkUtil
import org.apache.spark.rdd.JdbcRDD

/**
 * @Date 22.3.28
 * @Created by HANGGE
 * @Description
 *             通过JDBC创建RDD
 */
object C02_MakeRDDFromJDBC {

  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSc
    // 获取连接函数
    val  getConn = () => {
      DriverManager.getConnection("jdbc:mysql://localhost:3306/doit30", "root", "root")
    }
  // 处理结果数据的函数  注意返回值和JdbcRDD 的泛型一致
    val  optionResultSet = (rs:ResultSet)=>{
      val uid = rs.getInt(1)
      val name = rs.getString(2)
      val age = rs.getInt(3)
      val gender = rs.getString(4)
      val city = rs.getString(5)
      (uid,name , age ,gender , city)
    }


    /**
     * 1 泛型 返回结果的数据类型  case  class  元组
     * (
     * sc: SparkContext,  环境
     * getConnection: () => Connection,  连接函数
     * sql: String,  sql
     * lowerBound: Long,起始行  第一个问好
     * upperBound: Long, 结束行    第二个问好
     * numPartitions: Int,
     * mapRow: (ResultSet) => T = JdbcRDD.resultSetToObjectArray _)
     */
    val rdd: JdbcRDD[(Int, String, Int, String, String)] = new JdbcRDD[(Int, String, Int, String, String)](
      sc,
      getConn,
      "select * from tb_user where ? <= uid and  uid <=?",
        1, // 起始行  第一个问好
       4, //结束行    第二个问好
      2,  // 设置分区数
      optionResultSet
    )
    // 打印分区数
    println(rdd.getNumPartitions)
    // 执行行动算子打印结果
    rdd.foreach(println)
  }
}
